package com.ksc.wordcount.task.map;

import com.ksc.wordcount.datasourceapi.PartionFile;
import com.ksc.wordcount.datasourceapi.PartionReader;
import com.ksc.wordcount.conf.AppConfig;
import com.ksc.wordcount.shuffle.DirectShuffleWriter;
import com.ksc.wordcount.task.Task;
import com.ksc.wordcount.task.map.MapFunction;
import com.ksc.wordcount.task.map.MapStatus;
import com.ksc.wordcount.task.map.MapTaskContext;

import java.io.IOException;
import java.util.UUID;
import java.util.stream.Stream;

public class ShuffleMapTask extends Task<MapStatus> {

    PartionFile partiongFile;  // 分区文件
    PartionReader partionReader;  // 分区读取器
    int reduceTaskNum;  // Reduce任务数量
    MapFunction mapFunction;  // Map函数

    public ShuffleMapTask(MapTaskContext mapTaskContext) {
        super(mapTaskContext);
        this.partiongFile = mapTaskContext.getPartiongFile();
        this.partionReader = mapTaskContext.getPartionReader();
        this.reduceTaskNum = mapTaskContext.getReduceTaskNum();
        this.mapFunction = mapTaskContext.getMapFunction();
    }

    //写shuffle
    public MapStatus runTask() throws IOException {
        // 将分区文件转换为流
        Stream<String> stream = partionReader.toStream(partiongFile);

//        Stream<AbstractMap.SimpleEntry<String, Integer>> simpleEntryStream = stream.flatMap(line -> Arrays.stream(line.split("\\s+")))
//                .map(word -> new AbstractMap.SimpleEntry<String, Integer>(word, 1));

        // 使用Map函数对流进行处理
        Stream kvStream = mapFunction.map(stream);

        String shuffleId = UUID.randomUUID().toString();

        // 将任务执行结果写入Shuffle文件中
        DirectShuffleWriter shuffleWriter = new DirectShuffleWriter(AppConfig.shuffleTempDir, shuffleId, applicationId, partionId, reduceTaskNum);

        //6、executor写shuffle数据到本地
        shuffleWriter.write(kvStream);
        shuffleWriter.commit();

        // 返回Shuffle任务的状态
        return shuffleWriter.getMapStatus(taskId);
    }
}